package com.gitee.xhs.savePointReader.windowState;

import com.gitee.xhs.windowState.TransactionCounterDao;
import org.apache.flink.state.api.functions.WindowReaderFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/7/8 14:59
 * @description WindowStateReaderFunction
 */
public class WindowStateReaderFunction extends WindowReaderFunction<TransactionCounterDao, WindowStateExhibitor, Long, TimeWindow> {

    private final String jobId;
    private final String operatorUid;

    public WindowStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void readWindow(Long key, Context<TimeWindow> context, Iterable<TransactionCounterDao> elements, Collector<WindowStateExhibitor> out) throws Exception {
        TransactionCounterDao transactionCounterDao = elements.iterator().next();

        WindowStateExhibitor windowStateExhibitor = new WindowStateExhibitor();
        windowStateExhibitor.setJobId(jobId);
        windowStateExhibitor.setOperatorUid(operatorUid);
        windowStateExhibitor.setTimeWindow(context.window());
        windowStateExhibitor.setTriggerTimers(context.registeredEventTimeTimers());

        windowStateExhibitor.setAccountId(transactionCounterDao.getAccountId());
        windowStateExhibitor.setStartTime(transactionCounterDao.getStartTime());
        windowStateExhibitor.setEndTime(transactionCounterDao.getEndTime());
        windowStateExhibitor.setSumAmount(transactionCounterDao.getSumAmount());

        out.collect(windowStateExhibitor);
    }
}
